package flink.demo;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Map;

public class CustomSourceFunction {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStreamSource<MySourceFunction.EventLog> dataStreamSource = environment.addSource(new MySourceFunction());
        dataStreamSource.print();
        environment.execute();
    }
}


class MySourceFunction implements SourceFunction<MySourceFunction.EventLog> {
    volatile boolean flag = true;


    @Override
    public void run(SourceContext<EventLog> sourceContext) throws Exception {
        EventLog eventLog = new EventLog();
        String[] events = {"appLaunch", "pageLoad", "adShow", "adClick", "itemShare", "itemCollect", "putBack", "wakeUp", "appClose"};
        HashMap<String, String> eventInfoMap = new HashMap<>();

        while (flag) {

            eventLog.setGuid(RandomUtils.nextLong(1, 1000));
            eventLog.setSessionId(RandomStringUtils.randomAlphabetic(12).toUpperCase());
            eventLog.setTimeStamp(System.currentTimeMillis());
            eventLog.setEventId(events[RandomUtils.nextInt(0, events.length)]);

            eventInfoMap.put(RandomStringUtils.randomAlphabetic(1), RandomStringUtils.randomAlphabetic(2));
            eventLog.setEventInfo(eventInfoMap);

            sourceContext.collect(eventLog);

            eventInfoMap.clear();

            Thread.sleep(RandomUtils.nextInt(200, 1500));
        }


    }

    @Override
    public void cancel() {
        flag = false;
    }

    class EventLog {

        @Override
        public String toString() {
            return "EventLog{" +
                    "guid=" + guid +
                    ", sessionId='" + sessionId + '\'' +
                    ", eventId='" + eventId + '\'' +
                    ", timeStamp=" + timeStamp +
                    ", eventInfo=" + eventInfo +
                    '}';
        }

        private long guid;
        private String sessionId;
        private String eventId;
        private long timeStamp;
        private Map<String, String> eventInfo;

        public long getGuid() {
            return guid;
        }

        public void setGuid(long guid) {
            this.guid = guid;
        }

        public String getSessionId() {
            return sessionId;
        }

        public void setSessionId(String sessionId) {
            this.sessionId = sessionId;
        }

        public String getEventId() {
            return eventId;
        }

        public void setEventId(String eventId) {
            this.eventId = eventId;
        }

        public long getTimeStamp() {
            return timeStamp;
        }

        public void setTimeStamp(long timeStamp) {
            this.timeStamp = timeStamp;
        }

        public Map<String, String> getEventInfo() {
            return eventInfo;
        }

        public void setEventInfo(Map<String, String> eventInfo) {
            this.eventInfo = eventInfo;
        }

    }
}

class MyParallelSourceFunction implements ParallelSourceFunction{

    @Override
    public void run(SourceContext sourceContext) throws Exception {

    }

    @Override
    public void cancel() {

    }
}

class MyRichSourceFunction extends RichSourceFunction<MySourceFunction.EventLog> {

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext runtimeContext = getRuntimeContext();
        // 可以从运行时上下文中，取到本算子所属的 task 的task名
        String taskName = runtimeContext.getTaskName();
        // 可以从运行时上下文中，取到本算子所属的 subTask 的subTaskId
        int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
        super.open(parameters);
    }

    @Override
    public void run(SourceContext<MySourceFunction.EventLog> sourceContext) throws Exception {

    }

    @Override
    public void cancel() {

    }
}